// Ninf-G-XX, ver. 0.1
// Copyright 2011 National Institute of Advanced Industrial Science and Technology
// 
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// 
//     http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include <boost/foreach.hpp>
#include <algorithm>
#include "GrpcData.hpp"
#include "GrpcDataImpl.hpp"
#include "GrpcException.hpp"
#include "URI.hpp"
#include "Types.hpp"

#include "FpOutputStream.hpp"
#include "FpInputStream.hpp"
#include "ByteArrayOutputStream.hpp"
#include "ByteArrayInputStream.hpp"
#include "DataOutputStream.hpp"
#include "DataInputStream.hpp"
#include "ProtocolFactory.hh"
#include "Deleter.hpp"
#include "Arg.hpp"

using namespace ninf::data;
using namespace ninf::lib;

GrpcData::GrpcData(
    const char ** list_of_URI_input,
    const char ** list_of_URI_output,
    grpc_data_type_t variable_type,
    const size_t* variable_configuration,
    const grpc_data_mode_t * storage_modes):
  immediate(NULL),
  variable_type(variable_type)
  {
      if (list_of_URI_input != NULL)
        for (const char ** tmp = list_of_URI_input; *tmp != NULL; tmp++) {
          ninf::lib::URI uri(*tmp);
          if (!uri.sane)
            throw GRPC_EXCEPTION("MalformedURI", GRPC_MALFORMED_URI);
          inputURIs.push_back(uri);
        }
      if (list_of_URI_output != NULL)
        for (const char ** tmp = list_of_URI_output; *tmp != NULL; tmp++) {
          ninf::lib::URI uri(*tmp);
          if (!uri.sane)
            throw GRPC_EXCEPTION("MalformedURI", GRPC_MALFORMED_URI);
          outputURIs.push_back(uri);
        }
      if (variable_configuration != NULL)
        for (const size_t * tmp = variable_configuration; *tmp != 0; tmp++)
          this->variable_configuration.push_back(*tmp);

      if (storage_modes != NULL)
        for (const grpc_data_mode_t * tmp = storage_modes; *tmp != GRPC_END_LIST; tmp++)
          this->storage_modes.push_back(*tmp);
  }


/** serialize GrpcData metadata */
 void GrpcData::writeTo(ninf::lib::DataOutputStream & dos){
      // number of items in the list
      dos.writeInt(this->inputURIs.size());
      BOOST_FOREACH(ninf::lib::URI u, inputURIs){
        dos.writeString(u.toString().c_str());
      }
      dos.writeInt(this->outputURIs.size());
      BOOST_FOREACH(ninf::lib::URI u, outputURIs){
        dos.writeString(u.toString().c_str());
      }

      dos.writeInt((int)this->variable_type);
      dos.writeInt(this->variable_configuration.size());
      BOOST_FOREACH(int i, variable_configuration)
        dos.writeInt(i);

      dos.writeInt(this->storage_modes.size());
      BOOST_FOREACH(grpc_data_mode_t m, this->storage_modes)
        dos.writeInt((int)m);
    }

/**
 * read the meta from the stream
 */
 void GrpcData::readFrom(ninf::lib::DataInputStream & dis){
  int size;
  // number of items in the list
  size = dis.readInt();
  for (int i = 0; i < size; i++)
      inputURIs.push_back(ninf::lib::URI(dis.readString()));

  size = dis.readInt();
  for (int i = 0; i < size; i++)
      outputURIs.push_back(ninf::lib::URI(dis.readString()));

  variable_type = (grpc_data_type_t)dis.readInt();

  size = dis.readInt();
  for (int i = 0; i < size; i++)
      variable_configuration.push_back(dis.readInt());

  size = dis.readInt();
  for (int i = 0; i < size; i++)
      storage_modes.push_back((grpc_data_mode_t)dis.readInt());


}

void GrpcData::writeTo(const char * filename){
  FpOutputStream fop(filename);
  DataOutputStream dos(&fop);
  writeTo(dos);
}

void GrpcData::readFrom(const char * filename){
  FpInputStream fip(filename);
  DataInputStream dis(&fip);
  readFrom(dis);
}

void * GrpcData::getLocalMemPointer() {
  BOOST_FOREACH(ninf::lib::URI u, inputURIs)
  if (u.scheme == "memory" && u.authority == "localhost")
    return mappingManager.get(u.path);
  return NULL;
  }
void * GrpcData::getLocalOutMemPointer() {
  BOOST_FOREACH(ninf::lib::URI u, outputURIs)
    if (u.scheme == "memory" && u.authority == "localhost" )
      return mappingManager.get(u.path);
    return NULL;
  }

GrpcData
GrpcData::rfileMod(
      const std::string & infile,
      const std::string & outfile) {
   // copy original, add rfile if immediate  is true;

   GrpcData tmp;
   if (immediate != NULL) {
     tmp.inputURIs.push_back(ninf::lib::URI(std::string("rfile:///") + infile));
     tmp.outputURIs.push_back(ninf::lib::URI(std::string("rfile:///") + outfile));
   }

   // replace memory protocol with rfile
   BOOST_FOREACH(ninf::lib::URI u, inputURIs) {
     if (u.scheme == "memory")
       tmp.inputURIs.push_back(ninf::lib::URI(std::string("rfile:///") + infile));
     else
       tmp.inputURIs.push_back(ninf::lib::URI(u.toString()));
   }
   BOOST_FOREACH(ninf::lib::URI u, outputURIs) {
     if (u.scheme == "memory")
       tmp.outputURIs.push_back(URI(std::string("rfile:///") + outfile));
     else
       tmp.outputURIs.push_back(URI(u.toString()));
   }
   BOOST_FOREACH(grpc_data_mode_t m, storage_modes)
    tmp.storage_modes.push_back(m);
   BOOST_FOREACH(size_t size, variable_configuration)
    tmp.variable_configuration.push_back(size);
   tmp.variable_type = variable_type;
   return tmp;
}


void GrpcData::save(const char * uriStr){
  URI uri(uriStr);
  Protocol * protocol = ProtocolFactory::getProtocol(uri.scheme);
  Deleter<Protocol> _d(protocol);
  try {
    ByteArrayOutputStream fos(1000, 1000);
    DataOutputStream dos(&fos);
    writeTo(dos);
    protocol->uploadDataMemory(uri.toString(), fos.getBytes(), fos.length());
  } catch (ProtocolError e){
    throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
  } catch (ninf::lib::Exception e){
      throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
  }
}

void GrpcData::load(const char * uriStr){
  URI uri(uriStr);
  Protocol * protocol = ProtocolFactory::getProtocol(uri.scheme);
  Deleter<Protocol> _d(protocol);
  try {
    void * ptr;
    size_t size;
    protocol->downloadDataMemory(uri.toString(), &ptr, &size);
    ByteArrayInputStream bis((const char *)ptr, size);
    DataInputStream dos(&bis);
    readFrom(dos);
    free(ptr);

  } catch (ProtocolError e){
    throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
  } catch (ninf::lib::Exception e){
    throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
  }
}


/**
 * copy output uris into input uris.
 */
void GrpcData::update(){
  BOOST_FOREACH(URI uri, outputURIs) {
    if (std::find(inputURIs.begin(), inputURIs.end(), uri) == inputURIs.end())
      inputURIs.push_back(uri);
  }
}

void GrpcData::writeMemoryDataTo(void * programBuffer, void ** ptr,  int * size){
  Arg arg(this, programBuffer);
  this->immediate = programBuffer;
  ByteArrayOutputStream fos(1000, 1000, false);
  DataOutputStream dos(&fos);
  arg.writeData(dos);
  *ptr =  (void *)fos.getBytes();
  *size = fos.length();
}

void GrpcData::readDataFrom(void * programBuffer, void * ptr, int size){
  ninf::lib::Arg arg(this, programBuffer);
  this->immediate = programBuffer;
  ByteArrayInputStream bis((char *)ptr, size);
  DataInputStream dis(&bis);
  arg.readData(dis);
}


/**
 * transfer data
 */
void GrpcData::transfer(
  const char ** list_of_input_URI,
  const char ** list_of_output_URI,
  const grpc_data_mode_t * list_of_input_modes,
  const grpc_data_mode_t * list_of_output_modes){

  std::vector<URI> tmpInput;
  std::vector<URI> tmpOutput;

  /* merge lists */
  BOOST_FOREACH(URI uri, inputURIs)
    tmpInput.push_back(uri);
  BOOST_FOREACH(URI uri, outputURIs)
    tmpOutput.push_back(uri);

  if (list_of_input_URI != NULL)
    for (const char ** tmp = list_of_input_URI; *tmp != NULL; tmp++)
      tmpInput.push_back(URI(*tmp));
  if (list_of_output_URI != NULL)
    for (const char ** tmp = list_of_output_URI; *tmp != NULL; tmp++)
      tmpOutput.push_back(URI(*tmp));

  if (tmpInput.size() == 0)
    throw GRPC_EXCEPTION("input list is empty", GRPC_INVALID_ARGUMENT);
  URI uri = tmpInput.front();

  void * ptr;
  size_t size;
  {
    if (uri.scheme == "memory"){
      void * programBuffer = mappingManager.get(uri.path);
      writeMemoryDataTo(programBuffer, &ptr, (int *)&size);
    } else {
      Protocol * protocol = ProtocolFactory::getProtocol(uri.scheme);
      Deleter<Protocol> _d(protocol);
      try {
        protocol->downloadDataMemory(uri.toString(), &ptr, &size);
      } catch (ProtocolError e){
        throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
      }
    }
    Freer _f(ptr);
    BOOST_FOREACH(URI uriOut, tmpOutput) {
      if (uriOut.scheme == "memory"){
        void * programBuffer = mappingManager.get(uriOut.path);
        readDataFrom(programBuffer, ptr, size);
      } else {
        Protocol * protocol = ProtocolFactory::getProtocol(uriOut.scheme);
        Deleter<Protocol> _d(protocol);
        try {
          protocol->uploadDataMemory(uriOut.toString(), ptr, size);
        } catch (ProtocolError e){
          throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
        }
      }
    }
  }
}



# if 0
void GrpcData::save(const char * uriStr){
  void * ptr;
  if ((ptr = getLocalMemPointer()) != NULL) {
    Arg arg;
    fakeupArg(arg, ptr);

    URI uri(uriStr);
    Protocol * protocol = ProtocolFactory::getProtocol(uri.scheme);
    Deleter<Protocol> _d(protocol);

    ByteArrayOutputStream fos(1000, 1000);
    DataOutputStream dos(&fos);
    arg.writeData(dos);
    try {
      protocol->uploadDataMemory(uri.toString(), fos.getBytes(), fos.length());
    } catch (ProtocolError e){
      throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
    }

  } else {
      // have to perform URI to URI transfer, not implemented.
  }
}

void GrpcData::load(const char * uriStr){
  void * ptr;
  if ((ptr = getLocalOutMemPointer()) != NULL) {
    Arg arg;
    fakeupArg(arg, ptr);

    URI uri(uriStr);
    Protocol * protocol = ProtocolFactory::getProtocol(uri.scheme);
    Deleter<Protocol> _d(protocol);

    void * ptr;
    size_t size;
    try {
      protocol->downloadDataMemory(uri.toString(), &ptr, &size);
    } catch (ProtocolError e){
      throw GRPC_EXCEPTION(e.message, GRPC_OTHER_ERROR_CODE);
    }
    ByteArrayInputStream bis((const char *)ptr, size);
    //ByteArrayInputStream bis((const char *)ptr, 0);
    DataInputStream dos(&bis);
    arg.readData(dos);
    free(ptr);
  } else {
      // have to perform URI to URI transfer, not implemented.
  }
}
#endif

static
char ** createURIList(std::vector<ninf::lib::URI> & uris){
  char ** tmp = new char * [uris.size() + 1];
  int counter = 0;
  BOOST_FOREACH(ninf::lib::URI uri, uris)
    tmp[counter++] = strdup(uri.toString().c_str());
  tmp[counter++] = NULL;
  return tmp;
}

static
grpc_data_mode_t * createStorageModeList(std::vector<grpc_data_mode_t> & list) {
  grpc_data_mode_t * tmp = new grpc_data_mode_t[list.size() + 1];
  int counter = 0;
  BOOST_FOREACH(grpc_data_mode_t mode, list)
    tmp[counter++] = mode;
  tmp[counter] = GRPC_END_LIST;
  return tmp;
}

grpc_error_t GrpcData::getInfo(
    grpc_data_info_type_t info_tag,
    const char * server_name,
    char ** info){
  switch (info_tag) {
  case GRPC_HANDLE:
    return GRPC_NOT_SUPPORTED;
  case GRPC_INPUT_URI:
    *info = (char *)createURIList(inputURIs);
    return GRPC_NO_ERROR;
  case GRPC_OUTPUT_URI:
    *info = (char *)createURIList(outputURIs);
    return GRPC_NO_ERROR;
  case GRPC_MANAGEMENT_MODE:
    *info = (char *)createStorageModeList(storage_modes);
    return GRPC_NO_ERROR;
  case GRPC_SIZE:
    return GRPC_NOT_SUPPORTED;
  case GRPC_DATA_TYPE:
    *info = (char *)&variable_type;
    return GRPC_NO_ERROR;
  case GRPC_LOCATIONS_LIST:
    return GRPC_NOT_SUPPORTED;
    return GRPC_NO_ERROR;
  case GRPC_STATUS:
    return GRPC_NOT_SUPPORTED;
  case GRPC_COHERENT:
    return GRPC_NOT_SUPPORTED;
  default:
    return GRPC_INVALID_ARGUMENT;
  }
}

namespace ninf {
namespace data {

std::ostream& operator<<(std::ostream& os, const GrpcData & data){
  os << "inputURI:" << std::endl;
  BOOST_FOREACH(ninf::lib::URI v, data.inputURIs){
    os << "\t" << v << std::endl;
  }
  os << "outputURI:" << std::endl;
  BOOST_FOREACH(ninf::lib::URI v, data.outputURIs){
    os << "\t" << v << std::endl;
  }

  os << data.variable_type << std::endl;
  os << "configuration:[ ";
  BOOST_FOREACH(int i, data.variable_configuration)
    os << i << ", ";
  os << "]\n";

  os << "mode:[ ";
  BOOST_FOREACH(grpc_data_mode_t i, data.storage_modes)
    os << i << ", ";
  os << "]\n";

  return os;
}
}
}


